package com.dal.consumer.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Service;

@Service
@Slf4j
@RocketMQMessageListener(topic = "users", consumerGroup =  "${demo.rocketmq.consumer.group_push}",selectorExpression = "${demo.rocketmq.push_tag}",enableMsgTrace = true, customizedTraceTopic = "my-trace-topic")
public class RocketDome implements RocketMQListener<String>
//        , RocketMQPushConsumerLifecycleListener
{
    @Override
    public void onMessage(String s) {
//        获取消息，并在控制台打印（生产者一有消息就接收了）
        log.info("@@@@@接收信息："+s);
    }
//    @Override
//    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
//        defaultMQPushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
//
//            for (MessageExt messageExt : msgs) {
//                log.info("重试次数:" + messageExt.getReconsumeTimes());
//                log.info("key:" + messageExt.getKeys());
//                log.info("接受到的消息:" + new String(messageExt.getBody()));
//            }
//            log.info("{} Receive   msgs.size {} {} ",  Thread.currentThread().getName(),msgs.size(),msgs);
//            // 标记消息已经被成功消费
//            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//
//        });
//    }

}
